package com.tianzehua.grpc;

import com.tianzehua.proto.*;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.time.LocalTime;
import java.util.Iterator;

public class GrpcClient {

    public static void main(String[] args) throws  Exception{
        ManagedChannel managedChannel = ManagedChannelBuilder
                .forAddress("localhost",8899)
                .usePlaintext(true)
                .build();
        /* 同步的阻塞的grpc服务器 */
        StudentServiceGrpc.StudentServiceBlockingStub blockingStub = StudentServiceGrpc.newBlockingStub(managedChannel);
        /* 一部的非阻塞的服务器 */
        StudentServiceGrpc.StudentServiceStub stub = StudentServiceGrpc.newStub(managedChannel);

        MyResponse myResponse = blockingStub.getRealNameByUsername(MyRequest.newBuilder().setUsername("zhangsan").build());

        System.out.println(myResponse.getRealname());

        System.out.println("-----------------------------------------------------------");

        Iterator<StudentResponse> studentByAge = blockingStub.getStudentByAge(StudentRequest.newBuilder().setAge(20).build());
        while (studentByAge.hasNext()){
            StudentResponse studentResponse =studentByAge.next();
            System.out.println(studentResponse.getName() + ";" + studentResponse.getAge() + ";" + studentResponse.getCity());
        }

        System.out.println("-----------------------------------------------------------");


        /* 服务器的响应对象的一个回调 */
        StreamObserver<StudentResponseList> studentResponseListStreamObserver = new StreamObserver<StudentResponseList>() {
            @Override
            public void onNext(StudentResponseList value) {

                value.getStudentResponseList().forEach(studentResponse -> {
                    System.out.println(studentResponse.getName());
                    System.out.println(studentResponse.getAge());
                    System.out.println(studentResponse.getCity());
                    System.out.println("@@@@@@@@@@@@@");
                });
            }
            @Override
            public void onError(Throwable t) {
                System.out.println(t.getMessage());
            }
            @Override
            public void onCompleted() {
                System.out.println("completed!");
            }
        };


        /* rpc GetStudentWrapperByAges(stream StudentRequest) returns (StudentResponseList){} 传入一个流式的request，
        StudentRequest 是一个带有年龄的流式请求，返回一个ResponseList */
        StreamObserver<StudentRequest> studentRequestStreamObserver = stub.getStudentWrapperByAges(studentResponseListStreamObserver);
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(20).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(30).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(40).build());
        studentRequestStreamObserver.onNext(StudentRequest.newBuilder().setAge(50).build());

        studentRequestStreamObserver.onCompleted();
        Thread.sleep(1000);


        System.out.println("-----------------------------------------------------------");

        StreamObserver<StreamResquest> streamResquestStreamObserver =
                stub.biTalk(new StreamObserver<StreamResponse>() {
                                @Override
                                public void onNext(StreamResponse value) {
                                    System.out.println(value.getResponseInfo());
                                }

                                @Override
                                public void onError(Throwable t) {
                                    System.out.println(t.getMessage());
                                }

                                @Override
                                public void onCompleted() {
                                    System.out.println("completed!");
                                }
                            }

                );
        for (int i = 0; i < 10; i ++){
            streamResquestStreamObserver.onNext(StreamResquest.newBuilder().setRequestInfo(LocalTime.now().toString()).build());
            Thread.sleep(1000);
        }

    }
}
